#include "core/async_packet_socket_impl.h"

namespace vz {

AsyncPacketSocketImpl::AsyncPacketSocketImpl(AsyncSocket* socket,
    size_t max_packet_size)
  : socket_(socket)
  , insize_(max_packet_size)
  , inpos_(0)
  , outsize_(max_packet_size)
  , outpos_(0) {
  ASSERT(socket_.get() != NULL);

  inbuf_ = new char[insize_];
  outbuf_ = new char[outsize_];

  socket_->SignalConnectEvent.connect(
    this, &AsyncPacketSocketImpl::OnConnectEvent);
  socket_->SignalReadEvent.connect(
    this, &AsyncPacketSocketImpl::OnReadEvent);
  socket_->SignalWriteEvent.connect(
    this, &AsyncPacketSocketImpl::OnWriteEvent);
  socket_->SignalCloseEvent.connect(
    this, &AsyncPacketSocketImpl::OnCloseEvent);
}

AsyncPacketSocketImpl::~AsyncPacketSocketImpl() {
  delete[] inbuf_;
  delete[] outbuf_;
}

SocketAddress AsyncPacketSocketImpl::GetLocalAddress() const {
  return socket_->GetLocalAddress();
}

SocketAddress AsyncPacketSocketImpl::GetRemoteAddress() const {
  return socket_->GetRemoteAddress();
}

void AsyncPacketSocketImpl::SetError(int error) {
  return socket_->SetError(error);
}

int AsyncPacketSocketImpl::GetError() const {
  return socket_->GetError();
}

int AsyncPacketSocketImpl::Connect(const SocketAddress &address) {
  return socket_->Connect(address);
}

int AsyncPacketSocketImpl::SendPacket(int16 cmd,
                                      const void *data /*= NULL*/,
                                      uint32 data_size /*= 0*/) {
  size_t free_size = outsize_ - outpos_;
  size_t packet_size = data_size + sizeof(PacketHeader);

  if (packet_size > outsize_) {
    SetError(EMSGSIZE);
    return -1;
  } else if (packet_size > free_size) {
    SetError(EAGAIN);
    return -1;
  }

  PacketHeader packet_header;
  packet_header.v = 'V';
  packet_header.z = 'Z';
  packet_header.cmd = HostToNetwork16(cmd);
  packet_header.data_size = HostToNetwork32(data_size);

  AppendToOutBuffer(&packet_header, sizeof(PacketHeader));
  AppendToOutBuffer(data, data_size);

  return FlushOutBuffer();
}

int AsyncPacketSocketImpl::Close() {
  return socket_->Close();
}

AsyncPacketSocketImpl::Ptr AsyncPacketSocketImpl::SharedFromThis() {
  return AsyncPacketSocketImpl::Ptr(this);
}

void AsyncPacketSocketImpl::OnConnectEvent(AsyncSocket* socket) {
  ASSERT(socket_.get() == socket);

  SignalConnectEvent(SharedFromThis());
}

void AsyncPacketSocketImpl::OnReadEvent(AsyncSocket* socket) {
  ASSERT(socket_.get() == socket);

  int len = socket_->Recv(inbuf_ + inpos_, insize_ - inpos_);
  if (len < 0) {
    // TODO: Do something better like forwarding the error to the user.
    if (!socket_->IsBlocking()) {
      LOG(LS_ERROR) << "Recv() returned error: " << socket_->GetError();
    }
    return;
  }

  inpos_ += len;

  ProcessInput(inbuf_, &inpos_);

  if (inpos_ >= insize_) {
    LOG(LS_ERROR) << "input buffer overflow";
    ASSERT(false);
    inpos_ = 0;
  }
}

void AsyncPacketSocketImpl::OnWriteEvent(AsyncSocket* socket) {
  ASSERT(socket_.get() == socket);

  if (outpos_ > 0) {
    FlushOutBuffer();
  }

  if (outpos_ == 0) {
    SignalWriteEvent(SharedFromThis());
  }
}

void AsyncPacketSocketImpl::OnCloseEvent(AsyncSocket* socket, int error) {
  ASSERT(socket_.get() == socket);

  SignalCloseEvent(SharedFromThis(), error);
}

void AsyncPacketSocketImpl::ProcessInput(char * data, size_t* len) {
  SocketAddress remote_addr(GetRemoteAddress());

  while (true) {
    if (*len < sizeof(PacketHeader))
      return;

    PacketHeader *packet_header = (PacketHeader*)data;

    if (packet_header->v == 'V' && packet_header->z == 'Z') {
      uint16 cmd = NetworkToHost16(packet_header->cmd);
      uint32 data_size = NetworkToHost32(packet_header->data_size);
      if (*len < sizeof(PacketHeader) + data_size) {
        return;
      }

      SignalPacketEvent(SharedFromThis(), cmd,
                        data + sizeof(PacketHeader), data_size);

      *len -= sizeof(PacketHeader) + data_size;
      if (*len > 0) {
        memmove(data, data + sizeof(PacketHeader) + data_size, *len);
      }
    } else {
      return;
    }
  }
}

void AsyncPacketSocketImpl::AppendToOutBuffer(const void *data, size_t len) {
  ASSERT(outpos_ + len <= outsize_);
  memcpy(outbuf_ + outpos_, data, len);
  outpos_ += len;
}

int AsyncPacketSocketImpl::FlushOutBuffer() {
  int res = socket_->Send(outbuf_, outpos_);
  if (res <= 0) {
    return res;
  }

  if (static_cast<size_t>(res) <= outpos_) {
    outpos_ -= res;
  } else {
    ASSERT(false);
    return -1;
  }

  if (outpos_ > 0) {
    memmove(outbuf_, outbuf_ + res, outpos_);
  }
  return res;
}

}  // namespace vz